package io.kiki.sba.registry.server.metadata.multi.cluster.remote;


import com.google.common.collect.Sets;
import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.metaserver.MultiClusterSyncInfo;
import io.kiki.sba.registry.server.metadata.bootstrap.config.MultiClusterMetaServerConfig;
import io.kiki.sba.registry.server.shared.constant.CentralControlLeaderLearnModeEnum;
import io.kiki.sba.registry.server.shared.meta.AbstractCentralControlLeaderExchanger;
import io.kiki.sba.registry.store.api.meta.MultiClusterSyncRepository;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Setter
@Getter
public class RemoteClusterCentralControlExchanger extends AbstractCentralControlLeaderExchanger {

    private static final Logger logger = LoggerFactory.getLogger(RemoteClusterCentralControlExchanger.class);


    @Autowired
    private MultiClusterMetaServerConfig multiClusterMetaServerConfig;

    @Autowired
    private MultiClusterSyncRepository multiClusterSyncRepository;

    /**
     * <dataCenter, syncInfo>
     */
    private volatile Map<String, MultiClusterSyncInfo> dataCenterToMultiClusterSyncInfoMap = new ConcurrentHashMap<>();

    public RemoteClusterCentralControlExchanger() {
        super(DataExchanger.REMOTE_CLUSTER_META);
    }

    @Override
    protected CentralControlLeaderLearnModeEnum getMode() {
        return CentralControlLeaderLearnModeEnum.loadbalancer;
    }

    @Override
    public int getRpcTimeoutMillis() {
        return multiClusterMetaServerConfig.getRemoteClusterRpcTimeoutMillis();
    }

    @Override
    public int getServerPort() {
        return multiClusterMetaServerConfig.getRemoteMetaServerPort();
    }

    @Override
    protected Collection<ChannelHandler> getClientHandlers() {
        return Collections.emptyList();
    }


    /**
     * update and remove cluster infos according to db
     */
    public void refreshDataCenterToMultiClusterSyncInfoMap() {
        Set<MultiClusterSyncInfo> updates = multiClusterSyncRepository.queryLocalMultiClusterSyncInfoSet();
        Set<String> remoteDataCenters = updates.stream().map(MultiClusterSyncInfo::getRemoteDataCenter).collect(Collectors.toSet());
        Set<String> removes = Sets.difference(dataCenterToMultiClusterSyncInfoMap.keySet(), remoteDataCenters);

        synchronized (this) {
            for (MultiClusterSyncInfo multiClusterSyncInfo : updates) {
                dataCenterToMultiClusterSyncInfoMap.put(multiClusterSyncInfo.getRemoteDataCenter(), multiClusterSyncInfo);
                logger.info("[refreshClusterInfos]update remote: {} config.", multiClusterSyncInfo);
            }

            for (String remove : removes) {
                dataCenterToMultiClusterSyncInfoMap.remove(remove);
                removeLeader(remove);
                logger.info("[refreshClusterInfos]remove remote: {} config.", remove);
            }
        }
    }

    @Override
    protected Collection<String> getCentralControlServerDomains(String dataCenter) {
        MultiClusterSyncInfo multiClusterSyncInfo = dataCenterToMultiClusterSyncInfoMap.get(dataCenter);
        if (multiClusterSyncInfo == null) {
            throw new RuntimeException(String.format("dataCenter: {} meta domain config not exist.", dataCenter));
        }
        return Collections.singleton(multiClusterSyncInfo.getRemoteMetaAddress());
    }

    public synchronized Set<String> getAllRemoteClusters() {
        return new HashSet<>(dataCenterToMultiClusterSyncInfoMap.keySet());
    }
}
